/**
 * Copyright (C) 2024-2030 Super BilGen & 超亿时代信息技术股份有限公司.All Rights Reserved.
 */
package cn.com.yang.file;

/**
 * @Model cn.com.yang.thread.IndustrialGradeLineProcessorFixed
 * @Desc  TODO
 * @author yangyijin
 * @since 2025年5月22日-下午2:51:50
 */
import java.io.BufferedOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

public class IndustrialGradeLineProcessorFixed {

    // 配置参数
    private static final int CPU_CORES = Runtime.getRuntime().availableProcessors();
    private static final int INDEX_BUFFER_SIZE = 256 * 1024;  // 256KB索引缓冲区
    private static final int READ_BUFFER_SIZE = 2 * 1024 * 1024; // 2MB读取缓冲区
    private static final AtomicLong globalLineCounter = new AtomicLong(0);

    public static void main(String[] args) throws Exception {
    	 final String inputFile = "F:\\20gb_data.txt";
         final String indexFile = "F:\\20gb_index.idx";

        // 阶段1：构建精准行索引
        if (!Files.exists(Paths.get(indexFile))) {
            buildAccurateIndex(inputFile, indexFile);
        }

        // 阶段2：并行处理（修复行数丢失问题）
        processWithLineAccuracy(inputFile, indexFile);
        System.out.println("Total lines processed: " + globalLineCounter.get());
    }

    /** 修复点1：精准索引构建（支持所有换行符类型） */
    private static void buildAccurateIndex(String dataFile, String indexFile) 
            throws IOException {
        try (RandomAccessFile raf = new RandomAccessFile(dataFile, "r");
             FileChannel channel = raf.getChannel();
             DataOutputStream indexOut = new DataOutputStream(
                 new BufferedOutputStream(Files.newOutputStream(Paths.get(indexFile))))) {

            ByteBuffer buffer = ByteBuffer.allocate(INDEX_BUFFER_SIZE);
            long filePos = 0;
            boolean newLine = true;
            boolean lastCR = false; // 处理CRLF标记

            while (channel.read(buffer) != -1) {
                buffer.flip();
                while (buffer.hasRemaining()) {
                    byte b = buffer.get();

                    // 核心修复：支持所有换行格式
                    if (b == '\r') {
                        lastCR = true;
                        continue;
                    }

                    if (b == '\n') {
                        if (lastCR) {
                            // 处理Windows换行\r\n
                            indexOut.writeLong(filePos - 1); // 记录\r的位置
                        } else {
                            // 处理Unix换行\n
                            indexOut.writeLong(filePos);
                        }
                        newLine = true;
                    } else {
                        if (newLine) {
                            indexOut.writeLong(lastCR ? filePos - 1 : filePos);
                            newLine = false;
                        }
                    }

                    lastCR = (b == '\r');
                    filePos++;
                }
                buffer.clear();
            }

            // 修复点2：处理最后一行无换行的情况
            if (!newLine) {
                indexOut.writeLong(filePos);
            }
        }
    }

    /** 修复点3：精确分片处理 */
    private static void processWithLineAccuracy(String dataFile, String indexFile) 
            throws Exception {
        long totalLines = Files.size(Paths.get(indexFile)) / Long.BYTES;
        long linesPerThread = (totalLines + CPU_CORES - 1) / CPU_CORES;

        ExecutorService executor = Executors.newFixedThreadPool(CPU_CORES);
        CompletionService<Long> compService = new ExecutorCompletionService<>(executor);

        try (RandomAccessFile indexRaf = new RandomAccessFile(indexFile, "r")) {
            for (int i = 0; i < CPU_CORES; i++) {
                long startLine = i * linesPerThread;
                long endLine = Math.min(startLine + linesPerThread, totalLines);

                // 获取精确的字节范围
                long startByte = getIndexValue(indexRaf, startLine);
                long endByte = (endLine >= totalLines) ? 
                    Files.size(Paths.get(dataFile)) : getIndexValue(indexRaf, endLine);

                compService.submit(() -> 
                    processSegment(dataFile, startByte, endByte, startLine, endLine),
                    endLine - startLine
                );
            }
        }

        // 验证完整性
        long totalProcessed = 0;
        for (int i = 0; i < CPU_CORES; i++) {
            totalProcessed += compService.take().get();
        }
        executor.shutdown();

        if (totalProcessed != totalLines) {
            throw new IllegalStateException("Line count mismatch! Expected: " + 
                totalLines + " Actual: " + totalProcessed);
        }
    }

    private static long getIndexValue(RandomAccessFile indexRaf, long line) 
            throws IOException {
        indexRaf.seek(line * Long.BYTES);
        return indexRaf.readLong();
    }

    /** 修复点4：线程安全行处理 */
    private static long processSegment(String filePath, long start, long end, 
                                     long startLine, long endLine) {
        try (RandomAccessFile raf = new RandomAccessFile(filePath, "r");
             FileChannel channel = raf.getChannel()) {

            ByteBuffer buffer = ByteBuffer.allocate(READ_BUFFER_SIZE);
            channel.position(start);

            long currentLine = startLine;
            StringBuilder lineBuilder = new StringBuilder(1024);
            boolean skipLF = false; // 处理CRLF残留

            while (channel.position() < end && currentLine < endLine) {
                buffer.clear();
                int bytesRead = channel.read(buffer);
                if (bytesRead == -1) break;

                buffer.flip();
                while (buffer.hasRemaining() && currentLine < endLine) {
                    byte b = buffer.get();

                    // 修复点5：精准换行检测
                    if (skipLF && b == '\n') {
                        skipLF = false;
                        continue;
                    }

                    switch (b) {
                        case '\r':
                            commitLine(lineBuilder, currentLine++);
                            skipLF = true;
                            break;
                        case '\n':
                            commitLine(lineBuilder, currentLine++);
                            skipLF = false;
                            break;
                        default:
                            if (skipLF) {
                                commitLine(lineBuilder, currentLine++);
                                skipLF = false;
                            }
                            lineBuilder.append((char) b);
                    }
                }
            }

            // 处理最后未提交的行
            if (lineBuilder.length() > 0 && currentLine <= endLine) {
                commitLine(lineBuilder, currentLine++);
            }
            return currentLine - startLine;
        }catch(IOException ex) {
        	System.out.println("异常：{}"+ex);
        }
        return 0;
    }

    private static void commitLine(StringBuilder builder, long lineNum) {
        String line = builder.toString();
        builder.setLength(0);
        System.out.println("lineNum-"+lineNum +":line >" +line);
        // 业务处理逻辑
        globalLineCounter.incrementAndGet();

        // 示例：每百万行打印进度1_000_000
        if (lineNum % 100  == 0) {
            System.out.printf("Processed line %,d: %s...%n", 
                lineNum, line.substring(0, Math.min(50, line.length())));
        }
    }
}
